﻿using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Text;
using System.Threading.Tasks;

using Microsoft.ServiceBus.Messaging;

using Newtonsoft.Json;
// ReSharper disable LocalizableElement

namespace Eldert.IoT.Windows.Administration
{
    public class SimpleEventProcessor : IEventProcessor
    {
        private static int _messageCounter;

        public Task OpenAsync(PartitionContext context)
        {
            Console.WriteLine($"EventProcessor started.  Partition: {context.Lease.PartitionId}, Current offset: {context.Lease.Offset}");
            return Task.FromResult<object>(null);
        }

        public async Task ProcessEventsAsync(PartitionContext context, IEnumerable<EventData> messages)
        {
            try
            {
                foreach (var eventData in messages)
                {
                    _messageCounter++;

                    #region Receive data

                    var message = Encoding.UTF8.GetString(eventData.GetBytes());
                    EngineInformation engineInformation = null;

                    #endregion

                    #region Convert to EngineInformation object

                    // Try retrieving engine information object
                    try
                    {
                        engineInformation = JsonConvert.DeserializeObject<EngineInformation>(message);
                    }
                    catch (Exception)
                    {
                        Console.WriteLine("Could not deserialize as EngineInformation object");
                    }

                    #endregion

                    #region Write output

                    var output = new StringBuilder($"Message received: {message}{Environment.NewLine}Partition: {context.Lease.PartitionId}{Environment.NewLine}");

                    if (engineInformation != null)
                    {
                        output.Append(
                            $"{Environment.NewLine}Ship: {engineInformation.ShipName}{Environment.NewLine}Engine: {engineInformation.EngineName}{Environment.NewLine}Created: {engineInformation.CreatedDateTime}{Environment.NewLine}RPM: {engineInformation.RPM}{Environment.NewLine}Temperature: {engineInformation.Temperature}{Environment.NewLine}Warning: {engineInformation.Warning}{Environment.NewLine}Engine Warning: {engineInformation.GetWarningType()}{Environment.NewLine}");
                    }

                    Console.WriteLine(output.ToString());

                    #endregion
                }

                #region Create checkpoint

                /* 
                    It is critical to consider checkpointing in context - it would be unwise to checkpoint every message.  
                    The storage account used for checkpointing probably wouldn't handle this load, 
                    but more importantly checkpointing every single event is indicative of a queued messaging pattern for which a Service Bus Queue may be a better option than an Event Hub.  
                    The idea behind Event Hubs is that you will get at least once delivery at great scale.  
                    By making your downstream systems idempotent it is easy to recover from failures or restarts that result in the same events being received multiple times. 
                */
                if (_messageCounter > 10)
                {
                    Console.WriteLine("Checkpoint placed");
                    await context.CheckpointAsync();
                    _messageCounter = 0;
                }

                #endregion
            }
            catch (Exception exception)
            {
                Debug.WriteLine(exception.Message);
                throw;
            }
        }

        public async Task CloseAsync(PartitionContext context, CloseReason reason)
        {
            Console.WriteLine($"EventProcessor is shutting down: Partition: {context.Lease.PartitionId} Reason: {reason}");

            if (reason == CloseReason.Shutdown)
            {
                await context.CheckpointAsync();
            }
        }
    }
}